package main

import (
	"context"
	"fmt"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/plain"
	"os"
	"time"
)

const ip string = "120.46.139.204:29193"
const username string = "aerosatlink"
const password string = "aerosatlink"

func main() {

	dialer := &kafka.Dialer{
		ClientID:  "asl-aaa-sat-xxx1",
		Timeout:   10 * time.Second, // 网络连接超时
		DualStack: true,
		SASLMechanism: plain.Mechanism{
			Username: username,
			Password: password,
		}, // 认证配置
	}

	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  []string{ip},
		GroupID:  "asl-aaa-test-consumer1",
		Topic:    "B-8331T",
		Dialer:   dialer,
		MaxBytes: 10e6, // 10MB
	})

	// 指定5秒过期时间

	for {
		func() {
			start := time.Now()
			ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
			defer cancel()
			m, err := r.FetchMessage(ctx)
			if err != nil {
				fmt.Printf("timeout err [%s] : %s\n", time.Since(start).String(), err.Error())
				return
			}
			fmt.Printf("message at offset %d: %s = %s\n",
				m.Offset, string(m.Key), string(m.Value))

			if err = r.CommitMessages(context.Background(), m); err != nil {
				fmt.Printf("commit err: %s\n", err.Error())
			}
		}()
	}

	if err := r.Close(); err != nil {
		fmt.Printf("failed to close reader: %s", err)
		os.Exit(1)
	}

}
